package org.witsoft.localmq.impl;

import lombok.extern.slf4j.Slf4j;
import org.witsoft.localmq.FileManager;
import org.witsoft.localmq.FileReader;
import org.witsoft.localmq.model.Offset;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import org.witsoft.localmq.model.Message;

import java.io.EOFException;
import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @ClassName: MqFileReaderImpl
 * @Description:
 * @Author: Colin
 * @Date: 2020/12/26 1:25 AM
 * @Version 1.0
 **/
@Slf4j
@RequiredArgsConstructor
public class FileReaderImpl implements FileReader {

    @NonNull
    private FileManager fileManager;

    @Override
    public List<Message> read(List<Offset> offsets) throws IOException {
        List<Message> messageList = new ArrayList<>(offsets.size());
        Map<Integer, List<Offset>> groups = offsets.stream().collect(Collectors.groupingBy(Offset::getSegmentId));
        for (int segmentId : groups.keySet()) {
            File dataFile = fileManager.getSegmentDataFile(segmentId);
            if(!dataFile.exists()){
                // 数据不完整则返回，等待下次获取
                return Collections.emptyList();
            }
            File idxFile = fileManager.getSegmentIndexFile(segmentId);
            if(!idxFile.exists()){
                // 数据不完整则返回，等待下次获取
                return Collections.emptyList();
            }
            try(RandomAccessFile dataRaf = new RandomAccessFile(dataFile, "r")){
                try(RandomAccessFile idxRaf = new RandomAccessFile(idxFile, "r")){
                    for (Offset offset : groups.get(segmentId)) {
                        int position = offset.getPosition();
                        int idxPosIndex = position * 12  + 4;
                        if(idxPosIndex > idxRaf.length() - 8){
                            break;
                        }
                        try{
                            idxRaf.seek(idxPosIndex);
                            long pointer = idxRaf.readLong();
                            dataRaf.seek(pointer);
                            int length = dataRaf.readInt();
                            byte[] data = new byte[length];
                            dataRaf.read(data);
                            Message message = Message.builder()
                                    .offset(offset.getCurrentOffset())
                                    .segment(segmentId)
                                    .position(position)
                                    .data(data)
                                    .build();
                            messageList.add(message);
                        }catch (EOFException e){
                            log.warn("数据位置读取失败[{},{}]", segmentId , position);
                        }
                    }
                }
            }

        }
        return messageList;
    }
}
